package message

import (
	"encoding/binary"
	"sync/atomic"
	"time"

	"gitee.com/ameise84/e3dds/internal/log"
	"gitee.com/ameise84/e3dds/internal/message/packet"
	"gitee.com/ameise84/e3lock"
	"gitee.com/ameise84/e3log"
	"gitee.com/ameise84/e3pool/obj_pool"
	"gitee.com/ameise84/e3utils/bytes_buffer"
	"gitee.com/ameise84/e3utils/str_conv"
	jsoniter "github.com/json-iterator/go"
	"github.com/sourcegraph/conc/panics"
)

var (
	_gCallSlow = time.Duration(2) * time.Second
	_gStepSlow = time.Duration(500) * time.Millisecond
	_gMsgPool  = obj_pool.NewPool[*Msg](func() *Msg {
		return &Msg{
			_data: data{
				Body: make([]byte, 0, 1024),
			},
			step: make([]e3log.LogMarshaller, 0, 4),
		}
	})
	_gMsgCount atomic.Int64

	_gSlowMapLock     e3lock.SpinLock
	_gSlowMap         = map[string]time.Time{}
	_gTotalSlowMsgCnt uint64
)

func takeMessage() *Msg {
	m := _gMsgPool.Get()
	m.clean()
	m.takeTime = time.Now()
	m.lastTime = m.takeTime
	m.step = m.step[:0]
	m.step = append(m.step, &stepRecorder{
		call: "takeMessage",
	})
	return m
}

func freeMessage(m *Msg) {
	if m.IsCall() && !m.HasResponse() {
		log.GetLogger().Error("request not response").Object(m).Println()
	}
	checkSlow(m, time.Now())
	m.clean()
	_gMsgPool.Put(m)
}

func checkSlow(m *Msg, freeTime time.Time) {
	if _gCallSlow > 0 {
		takeTime := m.takeTime
		name := m._data.ProtocName
		total := freeTime.Sub(takeTime)
		if total > _gCallSlow {
			_gSlowMapLock.Lock()
			_gTotalSlowMsgCnt++
			n := _gTotalSlowMsgCnt
			ot, ok := _gSlowMap[name]
			_gSlowMap[name] = freeTime
			_gSlowMapLock.Unlock()
			if !ok || freeTime.Sub(ot).Seconds() > 1.0 {
				log.GetLogger().Warn("slow Msg").Str("protoc name", name).Float64("total cost second", 2, total.Seconds()).Objects(m.step...).Uint64("total", n).Println()
			}
		}
	}
}

type data struct {
	From       Source `json:"From,omitempty"` //请求调用链
	To         Source `json:"To,omitempty"`   //发送到对象
	MsgKind    Kind   `json:"mk,omitempty"`   //请求类型
	MsgId      uint64 `json:"id,omitempty"`   //请求id
	ProtocName string `json:"name,omitempty"` //协议名
	Body       []byte `json:"Body,omitempty"` //协议数据
}

func (ts *data) Clean() {
	ts.From.Clean()
	ts.To.Clean()
	ts.MsgKind = MsgNet
	ts.MsgId = 0
	ts.ProtocName = ""
	ts.Body = ts.Body[:0]
}

func (ts *data) Clone(src data) {
	ts.From = src.From
	ts.To = src.To
	ts.MsgKind = src.MsgKind
	ts.MsgId = src.MsgId
	ts.setMsg(src.ProtocName, src.Body)
}

func (ts *data) setMsg(protocName string, body []byte) {
	ts.ProtocName = protocName
	size := len(body)
	if cap(ts.Body) < size {
		ts.Body = make([]byte, size)
	} else {
		ts.Body = ts.Body[:size]
	}
	if size > 0 {
		copy(ts.Body, body)
	}
}

type Msg struct {
	_data      data
	ctx        any
	rspHand    RspHandler
	retryTimes int  //重试投递次数
	hasRsp     bool //是否已返回请求

	//下面的是辅助功能
	takeTime time.Time
	lastTime time.Time
	step     []e3log.LogMarshaller
	isToJson e3lock.Once
	json     string
}

func (ts *Msg) E3LogMarshall() string {
	return ts.ToJson()
}

func (ts *Msg) ResponseForward(rsp Response) {
	ts.ResponseMsg(rsp.GetProtocName(), rsp.GetBody())
	FreeMessage(rsp.(*Msg))
}

func (ts *Msg) GetKind() Kind {
	return ts._data.MsgKind
}

func (ts *Msg) GetFromActor() *Source {
	return &ts._data.From
}

func (ts *Msg) GetToActor() *Source {
	return &ts._data.To
}

func (ts *Msg) GetMsgID() uint64 {
	return ts._data.MsgId
}

func (ts *Msg) GetProtocName() string {
	return ts._data.ProtocName
}

func (ts *Msg) GetBody() []byte {
	return ts._data.Body
}

func (ts *Msg) ToJson() string {
	ts.isToJson.Do(func() {
		marshal, err := jsoniter.ConfigFastest.Marshal(ts._data)
		if err != nil {
			log.GetLogger().Error("Msg To json marshal failed").Err(err).Println()
		}
		ts.json = str_conv.ToString(marshal)
	})
	return ts.json
}

func (ts *Msg) Context() any {
	return ts.ctx
}

func (ts *Msg) ResponseMsg(protoName string, body []byte) {
	if ts.rspHand == nil {
		log.GetLogger().Error("Msg rsp is nil").Object(ts).Println()
	}
	rsp := TakeMessage()
	rsp.SetKind(MsgRsp)
	rsp.SetID(ts._data.MsgId)
	to := &ts._data.To
	from := &ts._data.From
	rsp.SetFromActor(to.Service, to.Session, to.Clerk)
	rsp.SetToActor(from.Service, from.Session, from.Clerk)
	rsp.SetMsg(protoName, body)
	ts.rspHand.OnResponse(rsp)
	FreeMessage(rsp)
	ts.hasRsp = true
}

func (ts *Msg) BuildForward(srvId string, ssid string, cid string) *Msg {
	newMsg := TakeMessage()
	newMsg.Clone(ts)
	from := &ts._data.From
	newMsg.SetFromActor(from.Service, from.Session, from.Clerk)
	newMsg.SetToActor(srvId, ssid, cid)
	newMsg.BindContext(ts.ctx)
	ts.ctx = nil
	return newMsg
}

func (ts *Msg) clean() {
	ts._data.Clean()
	ts.ctx = nil
	ts.rspHand = nil
	ts.retryTimes = 0
	ts.hasRsp = false
	ts.step = ts.step[:0]
	ts.isToJson.Reset()
	ts.json = ""

}

func (ts *Msg) CloneData(src *Msg) {
	ts._data.Clone(src._data)
}

func (ts *Msg) Clone(src *Msg) {
	ts._data.Clone(src._data)
	ts.takeTime = src.takeTime
	ts.lastTime = src.lastTime
	size := len(src.step)
	if cap(ts.step) < size {
		ts.step = make([]e3log.LogMarshaller, size)
	} else {
		ts.step = ts.step[:size]
	}
	if size > 0 {
		copy(ts.step, src.step)
	}
	ts.isToJson.Reset()
	ts.json = ""
}

func (ts *Msg) BindContext(ctx any) {
	ts.ctx = ctx
}

func (ts *Msg) SetMsg(protocName string, body []byte) {
	ts._data.setMsg(protocName, body)
	ts._data.ProtocName = protocName
	size := len(body)
	ts._data.Body = ts._data.Body[:0]
	if size > 0 {
		ts._data.Body = append(ts._data.Body, body[:size]...)
	}
}

func (ts *Msg) SetKind(msgKind Kind) {
	ts._data.MsgKind = msgKind
}

func (ts *Msg) SetID(msgId uint64) {
	ts._data.MsgId = msgId
}

func (ts *Msg) SetFromActor(s string, sid string, cid string) {
	ts._data.From.Service = s
	ts._data.From.Session = sid
	ts._data.From.Clerk = cid
}

func (ts *Msg) SetToActor(s string, sid string, cid string) {
	ts._data.To.Service = s
	ts._data.To.Session = sid
	ts._data.To.Clerk = cid
}

func (ts *Msg) IsCall() bool {
	switch ts._data.MsgKind {
	case MsgNet, MsgCall:
		return true
	default:
		return false
	}
}

func (ts *Msg) IsResponse() bool {
	return ts._data.MsgKind == MsgRsp
}

func (ts *Msg) IsToFix() bool {
	return ts._data.To.Session != ""
}

func (ts *Msg) ResponseError(err error) {
	if ts.rspHand == nil {
		log.GetLogger().Error("Msg ctx is nil").Object(ts).Println()
	}
	rsp := TakeMessage()
	rsp.SetKind(MsgRspError)
	rsp.SetID(ts._data.MsgId)
	to := &ts._data.To
	from := &ts._data.From
	rsp.SetFromActor(to.Service, to.Session, to.Clerk)
	rsp.SetToActor(from.Service, from.Session, from.Clerk)
	rsp.SetMsg("eng_err_code", str_conv.ToBytes(err.Error()))
	ts.rspHand.OnResponse(rsp)
	ts.hasRsp = true
	FreeMessage(rsp)
}

func (ts *Msg) HasResponse() bool {
	return ts.hasRsp
}

func (ts *Msg) SetRspHandler(rspHand RspHandler) {
	ts.rspHand = rspHand
}

func (ts *Msg) GetCreateTime() time.Time {
	return ts.takeTime
}

func (ts *Msg) MarshalBinary() (data []byte, err error) {
	b := bytes_buffer.NewShiftBuffer(1024, 0, true)
	packet.PackString(&b, binary.BigEndian, ts._data.From.Service)
	packet.PackString(&b, binary.BigEndian, ts._data.From.Session)
	packet.PackString(&b, binary.BigEndian, ts._data.From.Clerk)
	packet.PackString(&b, binary.BigEndian, ts._data.To.Service)
	packet.PackString(&b, binary.BigEndian, ts._data.To.Session)
	packet.PackString(&b, binary.BigEndian, ts._data.To.Clerk)
	packet.PackUint8(&b, binary.BigEndian, ts._data.MsgKind)
	packet.PackUint64(&b, binary.BigEndian, ts._data.MsgId)
	packet.PackString(&b, binary.BigEndian, ts._data.ProtocName)
	packet.PackBytes(&b, binary.BigEndian, ts._data.Body)
	return b.Fetch()
}

func (ts *Msg) UnmarshalBinary(bs []byte) error {
	b := bytes_buffer.Warp(bs)
	ts.clean()
	rs := panics.Try(func() {
		ts._data.From.Service = packet.UnPackString(b, binary.BigEndian)
		ts._data.From.Session = packet.UnPackString(b, binary.BigEndian)
		ts._data.From.Clerk = packet.UnPackString(b, binary.BigEndian)
		ts._data.To.Service = packet.UnPackString(b, binary.BigEndian)
		ts._data.To.Session = packet.UnPackString(b, binary.BigEndian)
		ts._data.To.Clerk = packet.UnPackString(b, binary.BigEndian)
		ts._data.MsgKind = packet.UnPackUint8(b, binary.BigEndian)
		ts._data.MsgId = packet.UnPackUint64(b, binary.BigEndian)
		ts._data.ProtocName = packet.UnPackString(b, binary.BigEndian)
		ts._data.Body = packet.UnPackBytes(b, binary.BigEndian)
	})
	err := rs.AsError()
	if err != nil {
		log.GetLogger().Error("rpc Msg unmarshal failed").Err(err).Println()
	}
	return err
}

func (ts *Msg) AddRetryTimes() (bool, time.Time) {
	if ts.retryTimes > 3 {
		return false, time.Time{}
	}
	ts.retryTimes++
	return true, time.Now().Add(1 * time.Second)
}
